package org.activequant.data.util.subscription;

import java.util.concurrent.Executor;

import org.activequant.core.domainmodel.InstrumentSpecification;
import org.activequant.core.domainmodel.data.TradeIndication;
import org.activequant.core.types.TimeFrame;
import org.activequant.data.retrieval.ISubscription;
import org.activequant.data.retrieval.ITradeIndicationSubscriptionSource;
import org.activequant.util.pattern.events.IEventListener;

/**
 * Proxy that wraps an instance of ITradeIndicationSubscriptionSource and translates all events
 * into thread controlled by Executor instance. Typically, Executor uses single-threaded
 * strategy, therefore this proxy will translate all events to a single thread.<br/>
 * TradeIndicationSubscriptionSourceExecutorProxy extends ExecutorSubscriptionSourceBase&lt;TradeIndication&gt; implements ITradeIndicationSubscriptionSource.<br/>
 * Holds the following associated variables:
 * <ul>
 * <li>source(ITradeIndicationSubscriptionSource)</li>
 * </ul>
 * Holds the following inherited variables:
 * <ul>
 * <li>executor(Executor)</li>
 * <li>activeBackends(Map&lt;Topic, Subscription&gt;)</li>
 * <li>activeSubscriptions(Queue&lt;SubscriptionDelegate&gt;)</li>
 * </ul>
 * <b>History:</b><br>
 *  - [03.02.2008] Created (Mike Kroutikov)<br>
 *
 *  @author Mike Kroutikov
 */
public class TradeIndicationSubscriptionSourceExecutorProxy extends ExecutorSubscriptionSourceBase<TradeIndication> implements ITradeIndicationSubscriptionSource {
	/**
	 * private final ITradeIndicationSubscriptionSource source;
	 */
	private final ITradeIndicationSubscriptionSource source;
	/**
	 * constructs a TradeIndicationSubscriptionSourceExecutorProxy (extends ExecutorSubscriptionSourceBase&lt;TradeIndication&gt; implements ITradeIndicationSubscriptionSource) using the given
	 * source(ITradeIndicationSubscriptionSource) to set its associated source(ITradeIndicationSubscriptionSource).<br/>
	 * Sets its inherited executor(Executor) with the given executor(Executor).
	 * @param source
	 * @param executor
	 */
	public TradeIndicationSubscriptionSourceExecutorProxy(ITradeIndicationSubscriptionSource source, Executor executor) {
		super(executor);
		this.source = source;
	}
	/**
	 * creates/subscribes an ISubscription&lt;TradeIndication&gt; for the given spec(InstrumentSpecification) and timeFrame(TimeFrame) using the associated source(ITradeIndicationSubscriptionSource).<br/>
	 * Then it creates and returns a ThreadSubscription(Subscription subclass) whose job is to fire each thrown event(TradeIndication) through the inherited executor(Executor) thread mechanism.
	 */
	@Override
	protected Subscription createSubscription(InstrumentSpecification spec, TimeFrame timeFrame) {
		try {
			final ISubscription<TradeIndication> subscription = source.subscribe(spec);
			
			return new ThreadSubscription() {
				/**
				 * final IEventListener&lt;TradeIndication&gt; listener = new IEventListener&lt;TradeIndication&gt;() {...}<br/>
				 * A listener whose job is fire each received event(TradeIndication) through the inherited executor(Executor).
				 */
				final IEventListener<TradeIndication> listener = new IEventListener<TradeIndication>() {
					public void eventFired(TradeIndication event) {
						fireEventInExecutorThread(event);
					}
				};

				@Override
				protected void handleActivate() throws Exception {
					subscription.addEventListener(listener);
					subscription.activate();
				}

				@Override
				protected void handleCancel() throws Exception {
					subscription.removeEventListener(listener);
					subscription.cancel();
				}
			};
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	/**
	 * Returns/creates a SubscriptionDelegate(ISubscription&lt;TradeIndication&gt;) for the given spec(InstrumentSpecification) and TIMEFRAME_1_TICK.
	 */
	public ISubscription<TradeIndication> subscribe(InstrumentSpecification spec) throws Exception {
		return super.subscribe(spec, TimeFrame.TIMEFRAME_1_TICK);
	}
}
